package org.geektimes.reactive.streams.simple;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.LinkedList;
import java.util.List;

/**
 * @FileName: DefaultPublisher
 * @Description: TODO
 * @Author: MaWenHao
 * @Created: 2021/3/30 - 下午11:09
 */
public class DefaultPublisher<T> implements Publisher<T> {

    private List<DefaultSubscriberWrapper<T>> subscriberWrappers = new LinkedList<>();

    @Override
    public void subscribe(Subscriber<? super T> s) {
        DefaultSubscription defaultSubscription = new DefaultSubscription();
        s.onSubscribe(defaultSubscription);

        subscriberWrappers.add(new DefaultSubscriberWrapper<>(s, defaultSubscription));
    }

    /**
     * 发布
     *
     * @param data
     */
    public void publish(T data) {
        subscriberWrappers.forEach(ReactiveUtils.accept(subscriber -> {
            DefaultSubscription subscription = subscriber.getSubscription();
            if (subscription.isCanceled()) {
                System.err.println("数据被忽略:" + data);
                return;
            }
            subscriber.onNext(data);
        }));
    }

    public void onComplete() {
        subscriberWrappers.forEach(subscriber -> {
            subscriber.onComplete();
        });
    }

    public static void main(String[] args) {
        DefaultPublisher<Integer> publisher = new DefaultPublisher<>();
        publisher.subscribe(new DefaultSubscriber<>());
        for (int i = 0; i < 5; i++) {
            publisher.publish(i);
        }
    }

}
